package com.pad.dpscn.daag.sdk.core.strategy.rocketmq;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.messaging.support.MessageBuilder;

import com.pad.dpscn.daag.sdk.core.strategy.MessageQueueStrategy;

/**
 * 
 * com.pad.dpscn.daag.sdk.core.strategy.rocketmq Description:
 * RocketMQ消息队列策略实现类，实现了MessageQueueStrategy接口。 用于处理RocketMQ消息队列的发送逻辑。
 * 
 * @author yangdezhao
 * @since 2024年3月11日
 * @version 1.0
 */
public class RocketMQMessageQueueStrategy implements MessageQueueStrategy {

	// 使用LogFactory创建日志记录器，用于记录日志信息
	private static final Log log = LogFactory.getLog(RocketMQMessageQueueStrategy.class);

	// RocketMQ模板对象，用于发送消息
	private RocketMQTemplate rocketMQTemplate;

	/**
	 * 构造函数，通过RocketMQTemplate对象初始化RocketMQMessageQueueStrategy实例。
	 * 
	 * @param rocketMQTemplate RocketMQ模板对象
	 */
	public RocketMQMessageQueueStrategy(RocketMQTemplate rocketMQTemplate) {
		super();
		this.rocketMQTemplate = rocketMQTemplate;
	}

	/**
	 * 实现MessageQueueStrategy接口的sendMessage方法，用于同步发送消息到指定topic。
	 * 
	 * @param topic   目标topic
	 * @param message 要发送的消息内容
	 */
	@Override
	public void sendMessage(String topic, String message) {
		// 不包含tag，只发送消息到指定topic
		try {
			// 使用RocketMQTemplate的send方法发送消息，通过MessageBuilder构建消息对象
			rocketMQTemplate.send(topic, MessageBuilder.withPayload(message).build());
		} catch (Exception e) {
			// 如果发送过程中出现异常，记录错误日志
			log.error("RocketMQ发送消息时发生错误", e);
		}
	}

	

	/**
	 * 实现MessageQueueStrategy接口的asyncSendMessage方法，但方法名存在误导。 实际上，该方法也是同步发送消息，而非异步发送。
	 * 
	 * @param topic   目标topic
	 * @param message 要发送的消息内容
	 */
	@Override
	public void asyncSendMessage(String topic, String message) {
		// 不包含tag，只发送消息到指定topic
		try {
			// 使用RocketMQTemplate的syncSend方法同步发送消息，通过MessageBuilder构建消息对象
			// 方法名asyncSendMessage与实际行为（同步发送）不符，可能是一个命名错误
			rocketMQTemplate.syncSend(topic, MessageBuilder.withPayload(message).build());
		} catch (Exception e) {
			// 如果发送过程中出现异常，记录错误日志
			log.error("RocketMQ同步发送消息时发生错误", e);
		}
	}

	@Override
	public void asyncSendAndReserveResult(String topic, String message) {
		rocketMQTemplate.asyncSend(topic, MessageBuilder.withPayload(message).build(), new SendCallback() {
			@Override
			public void onSuccess(SendResult sendResult) {
				// 成功回调
				log.info("回调结果: " + sendResult);
			}

			@Override
			public void onException(Throwable throwable) {
				// 异常回调
				log.error("asyncSendAndReserveResult()", throwable);
			}
		});

	}

}